#include <atomic>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <random>
#include <sstream>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>

#include "asycipc/api.hpp"
#include "asycipc/frame.hpp"
#include "asycipc/named_pipe.hpp"

namespace {

struct StressOptions {
  std::string pipe_name = R"(\\.\pipe\asycipc_stress)";
  std::size_t clients = 32;
  std::chrono::seconds duration{60};
  std::size_t payload_bytes = 256;
  std::chrono::milliseconds client_sleep{0};
  std::chrono::milliseconds maintenance_interval{50};
  std::size_t read_buffer_size = 4096;
  std::size_t worker_threads = 0;
};

bool StartsWith(const std::string& value, const std::string& prefix) {
  return value.rfind(prefix, 0) == 0;
}

uint16_t ReadUint16(const std::vector<uint8_t>& buffer, std::size_t offset) {
  return static_cast<uint16_t>(buffer[offset]) |
         static_cast<uint16_t>(buffer[offset + 1]) << 8;
}

uint32_t ReadUint32(const std::vector<uint8_t>& buffer, std::size_t offset) {
  return static_cast<uint32_t>(buffer[offset]) |
         static_cast<uint32_t>(buffer[offset + 1]) << 8 |
         static_cast<uint32_t>(buffer[offset + 2]) << 16 |
         static_cast<uint32_t>(buffer[offset + 3]) << 24;
}

bool ReadNextFrame(asycipc::PipeHandle& handle, std::vector<uint8_t>& buffer,
                   asycipc::Frame* out_frame,
                   std::atomic<bool>& stop_flag) {
  constexpr std::size_t kHeaderSize = 24;
  constexpr std::size_t kTrailerSize = 4;

  while (true) {
    if (buffer.size() >= kHeaderSize) {
      const uint32_t payload_length = ReadUint32(buffer, 14);
      const uint16_t ext_length = ReadUint16(buffer, 22);
      const std::size_t frame_size = kHeaderSize + ext_length + payload_length + kTrailerSize;
      if (buffer.size() >= frame_size) {
        std::vector<uint8_t> frame_data(buffer.begin(), buffer.begin() + frame_size);
        buffer.erase(buffer.begin(), buffer.begin() + frame_size);
        *out_frame = asycipc::Frame::Deserialize(frame_data);
        return true;
      }
    }

    if (stop_flag.load(std::memory_order_relaxed)) {
      return false;
    }

    std::vector<uint8_t> temp;
    try {
      temp = handle.Read();
    } catch (...) {
      return false;
    }
    if (temp.empty()) {
      return false;
    }
    buffer.insert(buffer.end(), temp.begin(), temp.end());
  }
}

StressOptions ParseArgs(int argc, char** argv) {
  StressOptions options;
  for (int i = 1; i < argc; ++i) {
    std::string arg = argv[i];
    auto next = [&]() -> std::string {
      if (i + 1 >= argc) {
        throw std::invalid_argument("missing value for " + arg);
      }
      return std::string(argv[++i]);
    };
    if (arg == "--pipe") {
      options.pipe_name = next();
    } else if (arg == "--clients") {
      options.clients = static_cast<std::size_t>(std::stoul(next()));
    } else if (arg == "--duration") {
      options.duration = std::chrono::seconds(std::stoul(next()));
    } else if (arg == "--payload") {
      options.payload_bytes = static_cast<std::size_t>(std::stoul(next()));
    } else if (arg == "--sleep-ms") {
      options.client_sleep = std::chrono::milliseconds(std::stoul(next()));
    } else if (arg == "--maintenance-ms") {
      options.maintenance_interval =
          std::chrono::milliseconds(std::stoul(next()));
    } else if (arg == "--read-buffer") {
      options.read_buffer_size = static_cast<std::size_t>(std::stoul(next()));
    } else if (arg == "--worker-threads") {
      options.worker_threads = static_cast<std::size_t>(std::stoul(next()));
    } else if (arg == "--help" || arg == "-h") {
      std::cout << "Usage: winpipe_stress [options]\n"
                   "  --pipe <name>             Named pipe (default \\.\\pipe\\asycipc_stress)\n"
                   "  --clients <n>             Concurrent clients (default 32)\n"
                   "  --duration <sec>          Duration in seconds (default 60)\n"
                   "  --payload <bytes>         Payload size per frame (default 256)\n"
                   "  --sleep-ms <ms>           Delay between sends per client\n"
                   "  --maintenance-ms <ms>     Server maintenance interval (default 50)\n"
                   "  --read-buffer <bytes>     IOCP read buffer size (default 4096)\n"
                   "  --worker-threads <n>      IOCP worker threads (default hw concurrency)\n";
      std::exit(0);
    } else {
      throw std::invalid_argument("unknown argument: " + arg);
    }
  }
  return options;
}

}  // namespace

int main(int argc, char** argv) {
  try {
    StressOptions options = ParseArgs(argc, argv);
    if (options.worker_threads == 0) {
      options.worker_threads = std::max<std::size_t>(1, std::thread::hardware_concurrency());
    }

    std::atomic<bool> stop{false};
    std::atomic<uint64_t> message_count{0};
    std::atomic<uint64_t> error_count{0};

    auto server = asycipc::PipeServer::Builder()
                       .Endpoint("stress")
                       .AllowApp("stress-client")
                       .OnMessage([&](asycipc::Channel& channel, const asycipc::Frame&) {
                         (void)channel;
                         message_count.fetch_add(1, std::memory_order_relaxed);
                         return std::vector<asycipc::Frame>{};
                       })
                       .OnError([&](const std::string&) {
                         error_count.fetch_add(1, std::memory_order_relaxed);
                       })
                       .Build();
    server.Start();

    auto listener = asycipc::CreateNamedPipeListener({options.pipe_name});
    auto* listener_ptr = listener.get();

    auto handshake = [&](asycipc::PipeServer& srv, asycipc::PipeHandle& handle,
                         asycipc::SessionContext& session) -> bool {
      if (stop.load(std::memory_order_relaxed)) {
        return false;
      }
      std::vector<uint8_t> request_bytes;
      try {
        request_bytes = handle.Read();
      } catch (const std::exception& ex) {
        srv.ReportError(std::string("handshake read failed: ") + ex.what());
        return false;
      }
      if (request_bytes.empty()) {
        srv.ReportError("handshake empty request");
        return false;
      }
      std::string request(request_bytes.begin(), request_bytes.end());
      if (!StartsWith(request, "HELLO")) {
        srv.ReportError("unexpected handshake payload");
        return false;
      }

      asycipc::HandshakeRequest hs_request;
      hs_request.app_id = "stress-client";
      hs_request.capabilities = {"load-test"};
      hs_request.heartbeat_interval_seconds = 5;

      try {
        srv.Accept(hs_request, session);
      } catch (const std::exception& ex) {
        srv.ReportError(std::string("handshake negotiation failed: ") + ex.what());
        return false;
      }

      const std::string ack = "READY\n";
      try {
        handle.Write(std::vector<uint8_t>(ack.begin(), ack.end()));
      } catch (const std::exception& ex) {
        srv.ReportError(std::string("handshake ack failed: ") + ex.what());
        return false;
      }
      return true;
    };

    std::thread server_thread([&]() {
      try {
        asycipc::ServeNamedPipe(
            server, *listener_ptr, handshake, options.maintenance_interval,
            [&]() { return !stop.load(std::memory_order_relaxed); },
            options.worker_threads, options.read_buffer_size);
      } catch (const std::exception& ex) {
        std::cerr << "ServeNamedPipe failed: " << ex.what() << "\n";
      }
    });

    const auto deadline = std::chrono::steady_clock::now() + options.duration;
    std::vector<std::thread> workers;
    workers.reserve(options.clients);

    std::atomic<bool> shutting_down{false};
    std::atomic<uint64_t> ack_count{0};
    std::atomic<uint64_t> client_errors{0};

    auto record_client_error = [&]() {
      if (!shutting_down.load(std::memory_order_relaxed)) {
        client_errors.fetch_add(1, std::memory_order_relaxed);
      }
    };

    auto client_worker = [&](std::size_t client_id) {
      asycipc::PipeHandlePtr handle;
      try {
        handle = asycipc::ConnectNamedPipeEndpoint(
            {options.pipe_name, std::chrono::milliseconds(5000),
             std::chrono::milliseconds(20)});
      } catch (...) {
        record_client_error();
        return;
      }

      const std::string handshake_payload =
          "HELLO " + std::to_string(client_id) + "\n";
      try {
        handle->Write(std::vector<uint8_t>(handshake_payload.begin(),
                                           handshake_payload.end()));
      } catch (...) {
        record_client_error();
        return;
      }

      std::string handshake_ack;
      while (handshake_ack.find("READY") == std::string::npos) {
        std::vector<uint8_t> chunk;
        try {
          chunk = handle->Read();
        } catch (...) {
          record_client_error();
          return;
        }
        if (chunk.empty()) {
          record_client_error();
          return;
        }
        handshake_ack.append(chunk.begin(), chunk.end());
      }

      std::vector<uint8_t> payload(options.payload_bytes);
      std::mt19937 rng(static_cast<uint32_t>(client_id));
      std::uniform_int_distribution<int> dist(0, 255);
      for (auto& b : payload) {
        b = static_cast<uint8_t>(dist(rng));
      }

      std::vector<uint8_t> read_buffer;
      uint64_t seq = 1;

      while (!stop.load(std::memory_order_relaxed)) {
        asycipc::Frame frame = asycipc::Frame::MakeData(
            seq, payload, {}, asycipc::kFrameFlagAckRequired);
        try {
          handle->Write(frame.Serialize());
        } catch (...) {
          record_client_error();
          break;
        }

        asycipc::Frame ack;
        if (!ReadNextFrame(*handle, read_buffer, &ack, stop)) {
          record_client_error();
          break;
        }
        if (ack.type != asycipc::FrameType::kAck) {
          record_client_error();
          break;
        }
        ack_count.fetch_add(1, std::memory_order_relaxed);
        ++seq;

        if (options.client_sleep.count() > 0) {
          std::this_thread::sleep_for(options.client_sleep);
        }

        if (std::chrono::steady_clock::now() >= deadline) {
          break;
        }
      }
    };

    for (std::size_t i = 0; i < options.clients; ++i) {
      workers.emplace_back(client_worker, i + 1);
    }

    while (std::chrono::steady_clock::now() < deadline) {
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }

    shutting_down.store(true, std::memory_order_relaxed);
    stop.store(true, std::memory_order_relaxed);

    try {
      asycipc::ConnectNamedPipeEndpoint(
          {options.pipe_name, std::chrono::milliseconds(100),
           std::chrono::milliseconds(10)});
    } catch (...) {
    }

    for (auto& worker : workers) {
      if (worker.joinable()) {
        worker.join();
      }
    }

    if (server_thread.joinable()) {
      server_thread.join();
    }

    auto& telemetry = server.telemetry();
    std::cout << "===== Stress Summary =====\n";
    std::cout << "Duration: " << options.duration.count() << "s\n";
    std::cout << "Clients: " << options.clients << "\n";
    std::cout << "Messages processed: " << message_count.load() << "\n";
    std::cout << "Client ACKs: " << ack_count.load() << "\n";
    std::cout << "Server errors: " << error_count.load() << "\n";
    std::cout << "Client errors: " << client_errors.load() << "\n";
    std::cout << "ipc_winpipe_accept_total="
              << telemetry.GetCounter("ipc_winpipe_accept_total") << "\n";
    std::cout << "ipc_winpipe_handshake_success_total="
              << telemetry.GetCounter("ipc_winpipe_handshake_success_total") << "\n";
    std::cout << "ipc_winpipe_handshake_failure_total="
              << telemetry.GetCounter("ipc_winpipe_handshake_failure_total") << "\n";
    std::cout << "ipc_winpipe_connection_open_total="
              << telemetry.GetCounter("ipc_winpipe_connection_open_total") << "\n";
    std::cout << "ipc_winpipe_disconnect_total="
              << telemetry.GetCounter("ipc_winpipe_disconnect_total") << "\n";
    std::cout << "ipc_winpipe_io_error_total="
              << telemetry.GetCounter("ipc_winpipe_io_error_total") << "\n";
    std::cout << "ipc_winpipe_read_bytes_total="
              << telemetry.GetCounter("ipc_winpipe_read_bytes_total") << "\n";
    std::cout << "ipc_winpipe_write_bytes_total="
              << telemetry.GetCounter("ipc_winpipe_write_bytes_total") << "\n";
    std::cout << "ipc_winpipe_pending_ops="
              << telemetry.GetGauge("ipc_winpipe_pending_ops") << "\n";
    std::cout << "ipc_winpipe_connections_active="
              << telemetry.GetGauge("ipc_winpipe_connections_active") << "\n";
    std::cout << "===========================\n";

    return 0;
  } catch (const std::exception& ex) {
    std::cerr << "Fatal: " << ex.what() << "\n";
    return 1;
  }
}










